package async;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class NonBlockingConnector {

	private final static Logger log = LoggerFactory.getLogger(NonBlockingConnector.class);
	
	private final Selector selector;
	private final Queue<SocketChannel> requests;
	
	//@Property
	private Executor executor;

	public static void main(String[] args) throws Exception {
		
		NonBlockingConnector c = new NonBlockingConnector();
		c.executor = Executors.newFixedThreadPool(5);
		c.start();
		
		SocketChannel google = c.connect("google.com", 80);
		SocketChannel apache = c.connect("apache.org", 80);
		
		Thread.sleep(2000);
		
		log.debug("Google: {}", google);
		log.debug("Apache: {}", apache);
		
	}
	
	public NonBlockingConnector() throws IOException {
		selector = Selector.open();
		requests = new ConcurrentLinkedQueue<SocketChannel>();
	}
	
	public SocketChannel connect(final String hostname, final int port) throws IOException {
		final SocketChannel socketChannel = SocketChannel.open();
		executor.execute(new Runnable() {
			public void run() {
				try {
					socketChannel.configureBlocking(false);
				} catch (IOException e) {
					throw new RuntimeException(e);
				}
				log.debug("Connecting {}:{}", hostname, port);
				boolean connected = false;
				try {
					connected = socketChannel.connect(new InetSocketAddress(hostname, port));
				} catch (IOException e) {
					throw new RuntimeException(e);
				}
				if(!connected) {
					log.debug("Connection pending {}", socketChannel);
					log.debug("Queue connection request {}", socketChannel);
					requests.offer(socketChannel);
					log.debug("Wakeup selector {}", selector);
					selector.wakeup();
				}
			}});
		return socketChannel;
	}
	
	public void start() throws IOException {
		executor.execute(new Runnable() {
			public void run() {
				while(true) {
					try {
						log.debug("Select {}", selector);
						selector.select();
					} catch (IOException e) {
						throw new RuntimeException(e);
					}
					Iterator<SelectionKey> keys = selector.selectedKeys().iterator();
					while(keys.hasNext()) {
						SelectionKey key = keys.next();
						log.debug("Selected key {}", key);
						if(key.isValid() && key.isConnectable()) {
							log.debug("Connectable key {}", key);
							SocketChannel socketChannel = (SocketChannel) key.channel();
							log.debug("Connect to {}", socketChannel);
							try {
								if(socketChannel.finishConnect()) {
									log.debug("Connection established {}", socketChannel);
									log.debug("Cancel key {}", key);
									key.cancel();
								}
							} catch (IOException e) {
								throw new RuntimeException(e);
							}
						}
					}
					while(!requests.isEmpty()) {
						SocketChannel socketChannel = requests.poll();
						log.debug("Register connection request in selector {}", socketChannel);
						try {
							socketChannel.register(selector, SelectionKey.OP_CONNECT);
						} catch (ClosedChannelException e) {
							throw new RuntimeException(e);
						}
					}
				}
			}});
	}

}
